package com.taotao.delayjob.core.utils;

import lombok.AccessLevel;
import lombok.NoArgsConstructor;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

/**
 * <p>
 *
 * </p>
 *
 * @author youbeiwuhuan
 * @version 1.0.0
 * @date 2023/11/10 11:44
 */
@NoArgsConstructor(access = AccessLevel.PRIVATE)
public class ConcurrentUtil {

    /**
     * 并行执行,runs中第一个抛的异常会导致执行结束并抛出该异常,无异常则都执行完了返回
     */
    public static <V> void runParallelStopFirstException(ThreadPoolExecutor threadPool, Runnable... runs) throws Throwable {
        Callable<V>[] callables = new Callable[runs.length];
        for (int i = 0; i < runs.length; i++) {
            final Runnable r = runs[i];
            callables[i] = new Callable<V>() {
                @Override
                public V call() throws Exception {
                    r.run();
                    return null;
                }
            };
        }
        supplyParallelStopFirstException(threadPool, callables);
    }

    /**
     * 并行执行,callables中第一个抛的异常会导致执行结束并抛出该异常,无异常则都执行完了返回各个结果
     */
    public static <V> List<V> supplyParallelStopFirstException(ThreadPoolExecutor threadPool, Callable<V>... callables) throws Throwable {

        final int MAX = callables.length;
        List<Future<V>> futures = new ArrayList<>(MAX);
        AtomicLong count = new AtomicLong(0);
        CountDownLatch latch = new CountDownLatch(1);
        AtomicInteger eIdx = new AtomicInteger(-1);

        for (int i = 0; i < MAX; i++) {
            final Callable<V> c = callables[i];
            final int index = i;
            Future<V> future = threadPool.submit(new Callable<V>() {
                @Override
                public V call() throws Exception {

                    try {
                        V v = c.call();
                        long curr = count.incrementAndGet();
                        if (curr >= MAX) {
                            latch.countDown();
                        }
                        return v;
                    } catch (Exception e) {
                        eIdx.set(index);
                        latch.countDown();
                        throw e;
                    }
                }
            });

            futures.add(future);
        }
        try {
            latch.await();
        } catch (InterruptedException e) {
        }

        List<V> result = new ArrayList<>(MAX);
        if (-1 == eIdx.intValue()) {
            // 无异常
            for (Future<V> f : futures) {
                result.add(f.get());
            }
        } else {
            // 有异常
            try {
                futures.get(eIdx.intValue()).get();
            } catch (ExecutionException e) {
                throw e.getCause();
            }
        }

        return result;
    }


    public static ThreadPoolExecutor newThreadPoolExecutor(int corePoolSize, int maximumPoolSize, String threadNamePreffix) {
        return newThreadPoolExecutor(corePoolSize, maximumPoolSize, 5 * 60, 1000,
                new ThreadPoolExecutor.CallerRunsPolicy(), threadNamePreffix);
    }

    public static ThreadPoolExecutor newThreadPoolExecutor(int corePoorSize, int maximumPoolSize, int keepAliveSeconds, int queueCapacity, String threadNamePreffix) {
        return newThreadPoolExecutor(corePoorSize, maximumPoolSize, keepAliveSeconds, queueCapacity,
                new ThreadPoolExecutor.CallerRunsPolicy(), threadNamePreffix);
    }

    public static ThreadPoolExecutor newThreadPoolExecutor(int corePoorSize, int maximumPoolSize, int keepAliveSeconds, int queueCapacity,
                                                           RejectedExecutionHandler rejectedExecutionHandler, String threadNamePreffix) {
        return new ThreadPoolExecutor(corePoorSize, maximumPoolSize, keepAliveSeconds,
                TimeUnit.SECONDS, new ArrayBlockingQueue<>(queueCapacity), new ThreadFactory() {
            private final AtomicLong number = new AtomicLong();

            @Override
            public Thread newThread(Runnable r) {
                Thread newThread = Executors.defaultThreadFactory().newThread(r);
                newThread.setName(threadNamePreffix + number.getAndIncrement());
                newThread.setDaemon(false);
                return newThread;
            }
        }, rejectedExecutionHandler);
    }

}
